package cn.doitedu.demo.join;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Temporal时态关联 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射mysql中的 “人脸识别行为记录表”
        // 此案例中，是充当  “版本表”
        tenv.executeSql(
                "create table face_action(\n" +
                "    uid bigint,\n" +
                "    event_id string,\n" +
                "    action_time bigint,\n" +
                "    rt1 as to_timestamp_ltz(action_time,3),\n" +
                "    watermark for rt1 as rt1 ,\n" +
                "    primary key (uid) not enforced\n" +
                ") WITH (                   \n" +
                "'connector' = 'mysql-cdc', \n" +
                "'hostname' = 'doitedu',    \n" +
                "'port' = '3306',           \n" +
                "'username' = 'root',       \n" +
                "'password' = 'root',       \n" +
                "'database-name' = 'flinktest',\n" +
                "'table-name' = 'face_rec_actions'\n" +
                ")");

        // 建表，映射kafka中的ss-2 ： 人脸识别操作的结果流
        tenv.executeSql("CREATE TABLE face_result        (\n" +
                "  uid BIGINT                                \n" +
                "  ,face_status int                           \n" +
                "  ,ts bigint                                 \n" +
                "  ,rt2 as to_timestamp_ltz(ts,3)             \n" +
                "  ,watermark for rt2 as rt2                  \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'a-2',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g2',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");


        tenv.executeSql("select\n" +
                "    face_action.uid,\n" +
                "    face_action.event_id,\n" +
                "    face_action.action_time,\n" +
                "    face_result.uid,\n" +
                "    face_result.face_status,\n" +
                "    face_result.ts\n" +
                "\n" +
                "from face_result \n" +  // 探测表
                "join face_action " + // 版本表
                "for system_time as of face_result.rt2 \n" +  // 指定探测表的 时间语义字段
                "on face_result.uid = face_action.uid").print();
    }
}
